package com.kasinf.sample.kafka.producer;

import com.kasinf.sample.kafka.message.TestMessages;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

/**
 * 消息生产者
 *
 * @Author : Kaiser
 * @Date: 2025/5/28  16:12
 */
@Slf4j
@Component
public class MessageProducer {
    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;
    @Resource
    @Qualifier("kafkaBoardTemplate")
    private KafkaTemplate<Object, Object> kafkaBoardTemplate;

    public void send(TestMessages testMessages) {
        log.info("发送消息：{}", testMessages.getMessage());
        // 默认是异步发送
//        kafkaTemplate.send("KS_TEST_TOPIC", testMessages);

        // 这样做是同步发送
//        try {
//            kafkaTemplate.send("KS_TEST_TOPIC", testMessages).get();
//        } catch (InterruptedException | ExecutionException e) {
//            throw new RuntimeException(e);
//        }

        // 以事务模式发送，确保消息发送成功
        kafkaTemplate.executeInTransaction(callback-> callback.send("KS_TEST_TOPIC", testMessages));
    }

    public void push(TestMessages testMessages) {
        log.info("广播消息：{}", testMessages.getMessage());
        kafkaBoardTemplate.send("KS_BOARD_TOPIC", testMessages);
    }

}
